package com.graduation.smartparkingplatform.Utils;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.core.annotation.Order;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.net.InetSocketAddress;
import java.util.List;

@Slf4j
@Component
@Order(2)
public class CanalClient implements ApplicationRunner {

  @Autowired private RedisTemplate redisTemplate;

  private static RedisTemplate redisTemplateTemp;

  @Value("${Mycanal.hostName}")
  private String hostName;

  @Value("${Mycanal.port}")
  private Integer port;

  @Value("${Mycanal.destination}")
  private String destination;

  @Value("${Mycanal.username}")
  private String username;

  @Value("${Mycanal.password}")
  private String password;

  @Value("${Mycanal.listenTable}")
  private String listenTable;

  // 修改序列化方式
  @Autowired(required = false)
  public void setRedisTemplate(RedisTemplate redisTemplate) {
    RedisSerializer stringSerializer = new StringRedisSerializer();
    redisTemplate.setKeySerializer(stringSerializer);
    redisTemplate.setValueSerializer(stringSerializer);
    redisTemplate.setHashKeySerializer(stringSerializer);
    redisTemplate.setHashValueSerializer(stringSerializer);
    this.redisTemplate = redisTemplate;
  }

  @PostConstruct
  public void init() {
    redisTemplateTemp = redisTemplate;
  }

  @Override
  public void run(ApplicationArguments args) throws Exception {
    CanalConnector connector =
        CanalConnectors.newSingleConnector(
            new InetSocketAddress(hostName, port), destination, username, password);
    int batchSize = 100;
    try {
      connector.connect();
      connector.subscribe(listenTable); // 指定监听单表
      connector.rollback();
      while (true) {
        // 获取指定数量的数据
        Message message = connector.getWithoutAck(batchSize);
        long batchId = message.getId();
        int size = message.getEntries().size();
        if (batchId == -1 || size == 0) {
          try {
            Thread.sleep(1000);
          } catch (InterruptedException e) {
            log.error("canal init error-" + e);
            e.printStackTrace();
          }
        } else {
          printEntry(message.getEntries());
        }
        // 提交确认
        connector.ack(batchId);
      }
    } finally {
      connector.disconnect();
    }
  }

  private static void printEntry(List<CanalEntry.Entry> entrys) {
    for (CanalEntry.Entry entry : entrys) {
      if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN
          || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
        continue;
      }
      CanalEntry.RowChange rowChage = null;
      try {
        rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
      } catch (Exception e) {
        log.error("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(), e);
        throw new RuntimeException(
            "ERROR ## parser of eromanga-event has an error , data:" + entry.toString(), e);
      }
      CanalEntry.EventType eventType = rowChage.getEventType();
      System.out.println(
          String.format(
              "================> binlog[%s:%s] , name[%s,%s] , eventType : %s",
              entry.getHeader().getLogfileName(),
              entry.getHeader().getLogfileOffset(),
              entry.getHeader().getSchemaName(),
              entry.getHeader().getTableName(),
              eventType));

      for (CanalEntry.RowData rowData : rowChage.getRowDatasList()) {
        if (eventType == CanalEntry.EventType.DELETE) {
          redisDelete(rowData.getBeforeColumnsList());
        } else if (eventType == CanalEntry.EventType.INSERT) {
          redisInsert(rowData.getAfterColumnsList());
        } else {
          log.info("-------> before");
          printColumn(rowData.getBeforeColumnsList());
          log.info("-------> after");
          redisUpdate(rowData.getAfterColumnsList());
        }
      }
    }
  }

  private static void printColumn(List<CanalEntry.Column> columns) {
    for (CanalEntry.Column column : columns) {
      log.info(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
    }
  }

  private static void redisInsert(List<CanalEntry.Column> columns) {
    try {
      JSONObject json = new JSONObject();
      for (CanalEntry.Column column : columns) {
        json.put(column.getName(), column.getValue());
      }
      if (columns.size() > 0) {
        redisTemplateTemp
            .opsForValue()
            .set("spaceNo_" + columns.get(1).getValue(), json.toJSONString());
      }
    } catch (Exception e) {
      log.error("canal redisInsert error-" + e);
      e.printStackTrace();
    }
  }

  private static void redisUpdate(List<CanalEntry.Column> columns) {
    try {
      JSONObject json = new JSONObject();
      for (CanalEntry.Column column : columns) {
        json.put(column.getName(), column.getValue());
      }
      if (columns.size() > 0) {
        redisTemplateTemp
            .opsForValue()
            .set("spaceNo_" + columns.get(1).getValue(), json.toJSONString());
      }
    } catch (Exception e) {
      log.error("canal redisUpdate error-" + e);
      e.printStackTrace();
    }
  }

  private static void redisDelete(List<CanalEntry.Column> columns) {
    try {
      JSONObject json = new JSONObject();
      for (CanalEntry.Column column : columns) {
        json.put(column.getName(), column.getValue());
      }
      if (columns.size() > 0) {
        redisTemplateTemp.delete("spaceNo_" + columns.get(1).getValue());
      }
    } catch (Exception e) {
      log.error("canal redisDelete error-" + e);
      e.printStackTrace();
    }
  }
}
